package com.rwind.watermark;

import com.rwind.model.Event;
import org.apache.flink.api.common.eventtime.*;

/**
 <pre>
 水位线的特性：
 ⚫ 水位线是插入到数据流中的一个标记，可以认为是一个特殊的数据
 ⚫ 水位线主要的内容是一个时间戳，用来表示当前事件时间的进展
 水位线是基于数据的时间戳生成的
 ⚫ 水位线的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进
 ⚫ 水位线可以通过设置延迟，来保证正确处理乱序数据
 ⚫ 一个水位线 Watermark(t)，表示在当前流中事件时间已经达到了时间戳 t, 这代表 t 之
 前的所有数据都到齐了，之后流中不会出现时间戳 t’ ≤ t 的数据
 水位线是 Flink 流处理中保证结果正确性的核心机制，它往往会跟窗口一起配合，完成对
 乱序数据的正确处理。
 </pre>
 */
public class CustomWatermarkStrategy implements WatermarkStrategy<Event> {

    /**
     * 主要负责按照既定的方式， 基于时间戳生成水位线。
     * @param context
     * @return
     */
    @Override
    public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
       return new WatermarkGenerator<Event>(){

           private Long delayTime = 5000L; // 延迟时间
           private Long maxTs = Long.MIN_VALUE + delayTime + 1L; // 观察到的最大时间戳

           /**
            * 每个事件（数据）到来都会调用的方法，它的参数有当前事件、时间戳，
            * 以及允许发出水位线的一个 WatermarkOutput，可以基于事件做各种操作
            * @param event
            * @param l
            * @param watermarkOutput
            */
           @Override
           public void onEvent(Event event, long l, WatermarkOutput watermarkOutput) {
               // 每来一条数据就调用一次
               maxTs = Math.max(event.timestamp, maxTs); // 更新最大时间戳
           }

           /**
            * onPeriodicEmit()里调用 output.emitWatermark()，就可以发出水位线了；这
            * 个方法由系统框架周期性地调用，默认 200ms 一次。所以水位线的时间戳是依赖当前已有数
            * 据的最大时间戳的（这里的实现与内置生成器类似，也是减去延迟时间再减 1），但具体什么
            * 时候生成与数据无关
            *
            * 周期性调用的方法，可以由 WatermarkOutput 发出水位线。周
            * 期时间为处理时间，可以调用环境配置的.setAutoWatermarkInterval()方法来设
            * 置，默认为200ms。
            *周期性生成水位线的间隔时间
            * env.getConfig().setAutoWatermarkInterval(60 *1000L);
            *
            * @param output
            */
           @Override
           public void onPeriodicEmit(WatermarkOutput output) {
               // 发射水位线，默认 200ms 调用一次
               output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
           }
       };
     }

    /**
     * TimestampAssigner：主要负责从流中数据元素的某个字段中提取时间戳，并分配给元素。时间戳的分配是生成水位线的基础。
     * @param context
     * @return
     */
    @Override
    public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        //return WatermarkStrategy.super.createTimestampAssigner(context);
        return new SerializableTimestampAssigner<Event>(){

            @Override
            public long extractTimestamp(Event event, long recordTimestamp) {
                return event.timestamp;
            }
        };
    }

    /**
     *
     */
    private static class PunctuatedWatermarkStrategy implements  WatermarkGenerator<Event>{

        /**
         * * 断点式水位线生成器（Punctuated Generator）
         *             * 断点式生成器会不停地检测 onEvent()中的事件，当发现带有水位线信息的特殊事件
         *             * 时， 就立即发出水位线。一般来说，断点式生成器不会通过 onPeriodicEmit()发出水位线
         * @param event
         * @param l
         * @param watermarkOutput
         */
        @Override
        public void onEvent(Event event, long l, WatermarkOutput watermarkOutput) {
// 只有在遇到特定的 itemId 时，才发出水位线
            if (event.userName.equals("Mary")) {
                watermarkOutput.emitWatermark(new Watermark(event.timestamp - 1));
            }
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
// 不需要做任何事情，因为我们在 onEvent 方法中发射了水位线
        }
    }
}
